package drds.binlog.database_driver.socket;

import drds.binlog.common.Authors;
import drds.common.Author;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.util.internal.OutOfDirectMemoryError;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SystemPropertyUtil;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.NotImplementedException;

import java.io.IOException;
import java.net.SocketAddress;

/**
 * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
 */
@Slf4j
@Author(name = Authors.LI_YANG)
public class NettySocketChannel implements SocketChannel
{


    private static final int WAIT_PERIOD = 10;                                                                   // milliseconds
    private static final int DEFAULT_INIT_BUFFER_SIZE = 1024 * 1024;                                                          // 1MB，默认初始缓存大小
    // 参考 mysql-connector-java-5.1.40.jar: com.mysql.jdbc.MysqlIO.maxThreeBytes
    // < 256 * 256 * 256 = 16MB
    private static final int DEFAULT_MAX_BUFFER_SIZE = 16 * DEFAULT_INIT_BUFFER_SIZE;                                        // 16MB，默认最大缓存大小
    private Channel channel = null;
    private Object lock = new Object();
    private ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE); // 缓存大小
    private int initMaxCapacity = byteBuf.maxCapacity();

    public Channel getChannel()
    {
        return channel;
    }

    //
    public void setChannel(Channel channel)
    {
        this.channel = channel;
    }
    //

    //
    public SocketAddress getLocalSocketAddress()
    {
        return channel != null ? channel.localAddress() : null;
    }

    public SocketAddress getRemoteSocketAddress()
    {
        return channel != null ? channel.remoteAddress() : null;
    }

    public boolean isConnected()
    {
        return channel != null ? true : false;
    }
    //


    public void write(@NonNull ByteBuf byteBuf) throws InterruptedException, IOException
    {
        synchronized (lock)
        {
            while (true)
            {
                if (null == this.byteBuf)
                {
                    throw new IOException("socket is closed !");
                }

                // source buffer is empty.
                if (!byteBuf.isReadable())
                {
                    break;
                }

                // 默认缓存大小不够用时需自动清理或扩充，否则将因缓存空间不足而造成I/O超时假象
                int fromBytesSize = byteBuf.readableBytes();
                int from$toBytesSize = fromBytesSize - this.byteBuf.writableBytes();
                if (from$toBytesSize > 0)
                {
                    // 首先避免频繁分配内存（扩容/收缩），其次避免频繁移动内存（清理）
                    if (this.byteBuf.readerIndex() >= from$toBytesSize)
                    { // 可以清理
                        // 回收已读空间，重置读写指针
                        this.byteBuf.discardReadBytes();
                        // 恢复自动扩充的过大缓存到默认初始缓存大小，释放空间
                        int oldCapacity = this.byteBuf.capacity();
                        if (oldCapacity > DEFAULT_MAX_BUFFER_SIZE)
                        { // 尝试收缩
                            int newCapacity = this.byteBuf.writerIndex();
                            newCapacity = ((newCapacity - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            int quarter = (newCapacity >> 2); // 至少留空四分之一
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            newCapacity += quarter; // 留空四分之一
                            if (newCapacity < (oldCapacity >> 1))
                            { // 至少收缩二分之一
                                try
                                {
                                    this.byteBuf.capacity(newCapacity);
                                    log.info("shrink byteBuf capacity: {} - {} = {} bytes", oldCapacity, oldCapacity - newCapacity, newCapacity);
                                } catch (OutOfMemoryError outOfMemoryError)
                                {
                                    initMaxCapacity = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                    log.warn("byteBuf OutOfMemoryError: {} bytes", newCapacity, outOfMemoryError);
                                }
                            }
                        }
                    } else
                    { // 尝试扩容
                        int oldCapacity = this.byteBuf.capacity();
                        if (oldCapacity < initMaxCapacity)
                        {
                            int quarter = (oldCapacity >> 2); // 至少扩容四分之一
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            from$toBytesSize = ((from$toBytesSize - 1) / quarter + 1) * quarter; // 对齐
                            int newCapacity = oldCapacity + from$toBytesSize;
                            if (newCapacity > initMaxCapacity)
                            {
                                newCapacity = initMaxCapacity;
                            }
                            try
                            {
                                this.byteBuf.capacity(newCapacity);
                                log.info("expand byteBuf capacity: {} + {} = {} bytes", oldCapacity, newCapacity - oldCapacity, newCapacity);
                            } catch (OutOfDirectMemoryError outOfDirectMemoryError)
                            {
                                // failed to allocate 885571168 byte(s) of
                                // direct memory (used: 1002946176, max:
                                // 1888485376)
                                long maxDirectMemory = SystemPropertyUtil.getLong("io.netty.maxDirectMemory", -1);
                                if (maxDirectMemory < 0)
                                {
                                    maxDirectMemory = PlatformDependent.maxDirectMemory();
                                }
                                if (initMaxCapacity > maxDirectMemory)
                                {
                                    initMaxCapacity = (int) maxDirectMemory;
                                    newCapacity = initMaxCapacity;
                                    log.warn("resize initMaxCapacity: {} bytes", initMaxCapacity, outOfDirectMemoryError);
                                    try
                                    {
                                        this.byteBuf.capacity(newCapacity);
                                        log.info("expand byteBuf capacity: {} + {} = {} bytes", oldCapacity, newCapacity - oldCapacity, newCapacity);
                                    } catch (OutOfMemoryError ignore)
                                    {
                                        initMaxCapacity = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                        log.warn("byteBuf OutOfMemoryError: {} bytes", newCapacity, ignore);
                                    }
                                } else
                                {
                                    initMaxCapacity = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                    log.warn("byteBuf OutOfDirectMemoryError: {} bytes", newCapacity, outOfDirectMemoryError);
                                }
                            } catch (OutOfMemoryError outOfMemoryError)
                            {
                                initMaxCapacity = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                log.warn("byteBuf OutOfMemoryError: {} bytes", newCapacity, outOfMemoryError);
                            }
                        }
                    }
                    from$toBytesSize = fromBytesSize - this.byteBuf.writableBytes();
                }

                if (from$toBytesSize != fromBytesSize)
                {
                    // deltaSize <= 0 可全部写入，deltaSize > 0 只能部分写入
                    if (from$toBytesSize <= 0)
                    {
                        this.byteBuf.writeBytes(byteBuf, fromBytesSize);
                        break;
                    } else
                    {
                        this.byteBuf.writeBytes(byteBuf, fromBytesSize - from$toBytesSize);
                    }
                }
                // dest buffer is full.
                lock.wait(WAIT_PERIOD);
                // 回收已读空间，重置读写指针
                this.byteBuf.discardReadBytes();
            }
        }
    }

    //
    public void write(byte[]... bytes) throws IOException
    {
        if (channel != null && channel.isWritable())
        {
            channel.writeAndFlush(Unpooled.copiedBuffer(bytes));
        } else
        {
            throw new IOException("write failed ! please checking !");
        }
    }

    public byte[] read(int readSize) throws IOException
    {
        return read(readSize, 0);
    }

    public byte[] read(int readSize, int timeout) throws IOException
    {
        int accumulatedWaitTime = 0;

        // 若读取内容较长，则自动扩充超时时间，以初始缓存大小为基准计算倍数
        if (timeout > 0 && readSize > DEFAULT_INIT_BUFFER_SIZE)
        {
            timeout *= (readSize / DEFAULT_INIT_BUFFER_SIZE + 1);
        }
        do
        {
            if (readSize > byteBuf.readableBytes())
            {
                if (null == channel)
                {
                    throw new IOException("socket has Interrupted !");
                }

                if (timeout > 0)
                {
                    accumulatedWaitTime += WAIT_PERIOD;
                    if (accumulatedWaitTime > timeout)
                    {
                        StringBuilder sb = new StringBuilder("socket read timeout occured !");
                        sb.append(" readSize = ").append(readSize);
                        sb.append(", readableBytes = ").append(byteBuf.readableBytes());
                        sb.append(", timeout = ").append(timeout);
                        throw new IOException(sb.toString());
                    }
                }

                synchronized (this)
                {
                    try
                    {
                        wait(WAIT_PERIOD);
                    } catch (InterruptedException e)
                    {
                        throw new IOException("socket has Interrupted !");
                    }
                }
            } else
            {
                byte[] bytes = new byte[readSize];
                synchronized (lock)
                {
                    byteBuf.readBytes(bytes);
                }
                return bytes;
            }
        } while (true);
    }

    @Override
    public void read(byte[] bytes, int offset, int length, int timeout) throws IOException
    {
        throw new NotImplementedException();
    }

    //
    public void close()
    {
        if (channel != null)
        {
            channel.close();
        }
        channel = null;
        // A fatal error has been detected by the Java Runtime Environment:
        // EXCEPTION_ACCESS_VIOLATION (0xc0000005)
        synchronized (lock)
        {
            byteBuf.discardReadBytes();// 回收已占用的内存
            byteBuf.release();// 释放整个内存
            byteBuf = null;
        }
    }


}
